package com.clown.networkFlowAnalysis

import java.net.URL

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.function.{AllWindowFunction, WindowFunction}
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment, _}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

// 定义uv输出统计的样例类
case class UvCount(windowEnd: Long, count: Long)

object UniqueVisitor {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    // env.setParallelism(1)
    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val source: URL = getClass.getResource("/UserBehavior.csv")
    val inputStream: DataStream[String] = env.readTextFile(source.getPath)

    // 转换成样例类类型，并提取时间戳和watermark
    val dataStream: DataStream[UserBehavior] = inputStream
      .map(data => {
        val arr = data.split(",")
        UserBehavior(arr(0).toLong, arr(1).toLong, arr(2).toLong, arr(3), arr(4).toLong)
      })
      .assignAscendingTimestamps(_.timestamp * 1000L)


    val uvStream = dataStream
      .filter(_.behavior.equals("pv"))
      .timeWindowAll(Time.hours(1)) // 直接不分组，基于dataStream，开1小时滚动窗口
      .apply(new UvCountResult())

    uvStream.print()
    env.execute("uv job")
  }
}

// 自定义实现全窗口函数，用一个set结构来保存所有的数据，进行自动去重
class UvCountResult() extends AllWindowFunction[UserBehavior, UvCount, TimeWindow] {
  override def apply(window: TimeWindow, input: Iterable[UserBehavior], out: Collector[UvCount]): Unit = {
    // 定义一个set，保存所有的userId
    var userIdSet = Set[Long]()
    // 遍历窗口中的所有数据，把userId添加到set中，自动去重
    input
      .foreach(userBehavior => userIdSet += userBehavior.userId)
    // 将set的size作为去重后的uv值输出
    out.collect(UvCount(window.getEnd, userIdSet.size))
  }
}